package com.yanggu.bigdata.realtime.app.function

import com.alibaba.fastjson.JSONObject
import com.yanggu.bigdata.realtime.utils.DimQueryUtil
import org.apache.flink.runtime.concurrent.Executors
import org.apache.flink.streaming.api.scala.async.{ResultFuture, RichAsyncFunction}

import scala.concurrent.{ExecutionContext, Future}
import scala.util.{Failure, Success}

trait DimJoinRichAsyncFunction[T] extends RichAsyncFunction[T, T] {

  private implicit lazy val executor: ExecutionContext = ExecutionContext.fromExecutor(Executors.directExecutor())

  /**
   * 抽象字段, 返回Hbase中的维度表名
   *
   * @return
   */
  def tableName: String

  /**
   * 通过输入数据获取维度表的主键
   *
   * @param input 事实宽表数据
   * @return
   */
  def getKey(input: T): String

  /**
   * 根据查询得到的维度表数据来补充事实表的字段
   * 专业术语叫使用维度表染色事实表
   *
   * @param input   事实宽表数据
   * @param dimInfo 维度表数据
   */
  def join(input: T, dimInfo: JSONObject): Unit

  def asyncInvoke(input: T, resultFuture: ResultFuture[T]): Unit = {
    //scala的Future语法, 需要学习下
    //https://docs.scala-lang.org/zh-cn/overviews/core/futures.html
    //和Java8的CompleteFuture的api类似
    Future[JSONObject] {
      //通过表名和主键, 查询维度表信息
      DimQueryUtil.getDimInfo(tableName = tableName, pkValue = getKey(input))
    }
    .onComplete {
      case Success(dimInfo) =>
        if (dimInfo != null) {
          join(input, dimInfo)
        }
        //向下游传递数据
        resultFuture.complete(Iterable(input))
      case Failure(exception) =>
        exception.printStackTrace()
        //向下游传递异常信息
        resultFuture.completeExceptionally(exception)
    }
  }

}
